package streaming

import java.io.{BufferedReader, InputStreamReader}
import java.net.{ConnectException, Socket}
import java.nio.charset.StandardCharsets

import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.receiver.Receiver

object CustomerReceiver {

  def main(args: Array[String]): Unit = {
    //1.初始化Spark配置信息
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("StreamWordCount")

    //2.初始化SparkStreamingContext
    val ssc = new StreamingContext(sparkConf, Seconds(5))

    //3.创建自定义receiver的Streaming
    val lineStream = ssc.receiverStream(new MyReceiver("node01", 9999))

    //4.将每一行数据做切分，形成一个个单词
    val wordStream = lineStream.flatMap(_.split("\t"))

    //5.将单词映射成元组（word,1）
    val wordAndOneStream = wordStream.map((_, 1))

    //6.将相同的单词次数做统计
    val wordAndCountStream = wordAndOneStream.reduceByKey(_ + _)

    //7.打印
    wordAndCountStream.print()

    //8.启动SparkStreamingContext
    ssc.start()
    ssc.awaitTermination()
  }


  class MyReceiver(host:String,port:Int) extends Receiver[String](StorageLevel.MEMORY_ONLY){

    private var socket : Socket = _

    override def onStart(): Unit = {
      new Thread("Socket Receiver"){
        override def run(): Unit = {
          receive()
        }
      }.start()
    }

    def receive(): Unit ={
      try{
        socket = new Socket(host,port)
        val reader = new BufferedReader(new InputStreamReader(socket.getInputStream,StandardCharsets.UTF_8))

        var input : String = null

//        while((input = reader.readLine()) != null){
//          store(input)
//        }
      } catch {
        case e: ConnectException =>
          restart(s"Error connecting to $host:$port", e)
          return
      }
    }


    override def onStop(): Unit = {
      if(socket != null ){
        socket.close()
        socket = null
      }
    }
  }
}
